import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import java.util.Hashtable;
import java.util.Enumeration;

public class Step1 {
	
	

	public static class step1_mapper extends Mapper<Object, Text, Text, Text>{
	
		private Text outputkey = new Text();
		private Text outputvalue = new Text();
		
		/** Input: (<obj>, <"id	url|zeta|k|node_1|..|node_n">)
		 *  where 	  obj: 	     not significant
		 * 		  id: 	     id of a node (it's an hash), let's call it j
		 * 		  url: 	     url (or string identifier) of j
		 * 		  zeta:	     # of received visits to j
		 * 		  k:	     # of live coupons
		 * 		  node_1..n: adjacency list 
		 * 	
		 *  Output: (<id>, <"+N">) where N is # of visits to 'id' or
		 *  		(<id>, <"-url|zeta|node_1|..|node_n">)
		 *	    The char [+-] is used by the reducer for distinguish between
	         *	    the two types of records.
		 *  Note: The InputFormat KeyValueTextInputFormat isn't no more
		 *  available: the TextInputFormat must be used doing the split manually. 
		 */
		public void map(Object key, Text value, Context context) 
				throws IOException, InterruptedException{
			int zeta, k, neighbour, deads=0;
			float epsilon;
			String url, myinfo, tokens[], values[];
			Hashtable<Integer, Integer> cinfo = new Hashtable<Integer, Integer>();
	
			/*Input file's lines are in the form: id\turl|zeta|k|node_1|..|node_n
			 * tokens[0] = id
			 * tokens[1] = url|zeta|k|node_1|..|node_n
			 */
			tokens = value.toString().split("\t");
			values = tokens[1].split("[|]");
			url = values[0];
			zeta = Integer.parseInt(values[1]);
			k = Integer.parseInt(values[2]); //coupon attivi
			epsilon = context.getConfiguration().getFloat("epsilon", 0);
			
			
			
			//outputvalue.set("");

			
			//If the node has not adjacency list then kill all its coupons.
			if (values.length<=3) deads=k; 
			else{ //iterate over each live coupon:
				for (int i=0; i<k; i++){
					//Probability of reset:
					if (Math.random()>epsilon){ //The coupons is survived:
						/*Randomly select a neighbour. In 'values' array the neighbours are 
						 * in the range [3, values.length[ */
						neighbour = (int) (3 + Math.round(Math.random()*(values.length - 4)));
						
						//outputkey.set(values[neighbour]);
						//context.write(outputkey, outputvalue);

						int tmp = 0;

						//If the hashtable (cinfo) already contains this neighbour than
						//increment the associated value, else initialize it to 1
						if (cinfo.containsKey(neighbour)) tmp=cinfo.get(neighbour);
						cinfo.put(neighbour, tmp+1); 

					}else deads++; //The coupon is killed: increase the local deads counter
				}
			}

			//update the global deads counter
			context.getCounter(PR_COUNTERS.DEADS).increment(deads);
			
			//Scan the hashtable cinfo: for each key (which is a node id) send the number of visits to that node.
			Enumeration<Integer> kenum = cinfo.keys();
			while (kenum.hasMoreElements()){
				int ck = kenum.nextElement();
				outputkey.set(values[ck]);

				//Set the value equal to the number of visits to the node values[ck]
				outputvalue.set("+" + cinfo.get(ck));

				context.write(outputkey, outputvalue);
			
			}
			
			//Build myinfo value:
			myinfo = url + "|" + zeta + "|";
			for (int i=3; i<values.length; i++) myinfo += values[i] + "|";
			
			//Send it to node tokens[0] itself
			outputkey.set(tokens[0]);
			outputvalue.set("-" + myinfo);

			//System.out.println("MYINFO: outputkey: " + outputkey.toString() + "; outputvalue: " + outputvalue.toString() + "; url: " + url);
			context.write(outputkey, outputvalue);
		}
	}
	/*
	public static class step1_combiner extends Reducer<Text, Text, Text, Text> {
		Text output_value = new Text();

		public void combiner(Text id_nodo, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			Iterator<Text> it = values.iterator();
			int k = 0;
			while (it.hasNext()) {
				Text value = it.next();
				//if it is found an empty value, it increment a counter
				if ((value.toString()).charAt(0)=='-')
					k++;
				else {
					//otherwise the value is written as is it
					//because it represents the key used to describe the graph node with its own adjacently list
					context.write(id_nodo, value);
					//return;
				}
			}
			output_value.set("+"+k);
			System.out.println("COMBINER: id_nodo: " + id_nodo + "; k: " + k);
			context.write(id_nodo, output_value);
		}
	}
	
	*/


	/** Input: (<id>, <+N>) or
	 * 		   (<id>, <"id	url|zeta|node_1|..|node_n">)
	 *  If the first char of the value is '+' then the following integer is added to the number
	 *  of received visits, otherwise the value contains the information of the node (they will be used to
	 *  rebuild the file for the next iteration).
	 * 
	 *  Output: (<id>, <"url|zeta|k|node_1|..|node_n">)
	 */
	public static class step1_reducer extends Reducer<Text, Text, Text, Text> {
		Text output_value= new Text();

		public void reduce(Text id_nodo, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			Iterator<Text> it = values.iterator();
			long k = 0; //number of received visits
			String adj_list="";
			String myinfo[] = null;
			int tmpc = 0;
			String old = null, cur=null;


			//Iterate on each value
			while (it.hasNext()) {
				Text value=it.next();
				
				/*If the value=="" the increase k. k is the received visits to
				  the node <id> in this iteration and also the number of lives 
				  coupons for that node in the next iteration */
				
				//Check for bad records: it should never happen, just for debug purposes
				if (value.toString().length()<=0) {
					//throw new IOException("Empty string: key: " + id_nodo.toString());
					System.err.println("Empty string: key: " + id_nodo.toString());
				}else{
					String cval = value.toString().substring(1); //cut off the first character
					
					//If the first char is '+' increment k of the following integer
					if (value.toString().charAt(0)=='+'){ 
						k += Integer.parseInt(cval);
					}else{ //otw the record is the node's info.
						//tmpc++;

						//Store node's information in myinfo array
						myinfo=cval.split("[|]");

						//old=cur;
						//cur=cval;
					}
				}
				
				
			
				
			
			}
			
			
			//If myinfo is null it means that no myinfo record was received: the current node is a node which
			//receive visits but is not in the graph (is an external link). All the coupons that are sent
			//to this node will be killed.
			if (myinfo==null){
				//System.out.println("Warning: missing page: " + id_nodo.toString());
				context.getCounter(PR_COUNTERS.DEADS).increment(k);
				return;
			}
			
			//if (myinfo.length<2) System.out.println("ERROR: key: " + id_nodo.toString() + "; tmpc: " + tmpc + "; old: " + old + "; cur: " + cur); 
			//Update the global counter of survived coupons.
			context.getCounter(PR_COUNTERS.CURRENT).increment(k);
			
			//Sum the received visits in this iteration (k) to the number of 
			//visits received in all iterations (actual_z). Note that k is also the
			//number of live coupons for this node in the next iteration.
			int actual_zeta=Integer.parseInt(myinfo[1]);
			actual_zeta+=k;
			
			//re-build the adjacency list
			for(int i=2; i<myinfo.length;++i)
				adj_list += myinfo[i]+'|';
			
			String output_str= myinfo[0]+'|'+actual_zeta+'|'+k+'|'+adj_list;
			
			//Send out the key-value pair
			output_value.set(output_str);
			context.write(id_nodo, output_value);
		}
	}
}
